const amqp = require("amqplib");

async function consume(queue) {
  const connection = await amqp.connect("amqp://rabbitmq");
  const channel = await connection.createChannel();

  await channel.assertQueue(queue, { durable: false });
  await channel.assertExchange("hello", "direct", { durable: true });
  await channel.bindQueue(queue, "hello", "");

  console.log(`Waiting for messages in ${queue}`);

  channel.consume(queue, (msg) => {
    if (msg !== null) {
      console.log(`Simple Consumer Received: ${msg.content.toString()}`);
      channel.ack(msg);
    }
  });
}

async function consumeRoutingKey(exchange, exchangeType, ...routingKeys) {
  const connection = await amqp.connect("amqp://rabbitmq");
  const channel = await connection.createChannel();

  // 确保交换机存在
  await channel.assertExchange(exchange, exchangeType, { durable: false });

  const bindings = routingKeys.map(async (routingKey) => {
    // 声明一个临时队列
    const queue = await channel.assertQueue("", { exclusive: true });

    // 绑定队列到交换机
    console.log(
      `Binding Queue ${queue.queue} to ${exchangeType} Exchange:${exchange}, RoutingKey ${routingKey}`
    );
    await channel.bindQueue(queue.queue, exchange, routingKey);

    // 消费消息
    channel.consume(queue.queue, (msg) => {
      if (msg !== null) {
        console.log(
          `Exchange:${exchange}, RoutingKey:${routingKey}, Received: ${msg.content.toString()}`
        );
        channel.ack(msg);
      }
    });
  });

  await Promise.all(bindings);
}

async function fanoutConsumer(exchange, consumerCount = 1) {
  const connection = await amqp.connect("amqp://rabbitmq");
  const channel = await connection.createChannel();

  await channel.assertExchange(exchange, "fanout", { durable: false });

  const queues = [];
  for (let i = 0; i < consumerCount; i++) {
    const queue = await channel.assertQueue("", { exclusive: true });
    queues.push(queue.queue);
    channel.bindQueue(queue.queue, exchange, "");
  }

  console.log(`Waiting for messages from Fanout Exchange:${exchange}`);

  queues.forEach((queue) => {
    channel.consume(queue, (msg) => {
      if (msg !== null) {
        console.log(
          `Fanout Exchange:${exchange}, Received: ${msg.content.toString()}`
        );
        channel.ack(msg);
      }
    });
  });
}

async function main() {
  // 简易队列的消费者
  await consume("simple-hello-queue");

  // direct 交换机，绑定不同的 routingKey
  await consumeRoutingKey(
    "hello-direct-exchange",
    "direct",
    "key1",
    "key2",
    "key3"
  );

  // topic 交换机，绑定不同的 routingKey
  await consumeRoutingKey(
    "hello-topic-exchange",
    "topic",
    "key1.*",
    "key2.*",
    "key1.#"
  );

  // fanout 交换机，绑定不同的 routingKey
  fanoutConsumer("hello-fanout-exchange", 3);
}

main();
